

package org.example.flinksql.connector;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;


/**
 * Mysql CDC Connector demo
 */
public class Cdc {

	public static void main(String[] args) throws Exception {
		EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
		TableEnvironment tbEnv = TableEnvironment.create(bsSettings);

		tbEnv.executeSql("CREATE TABLE t_logs (" +
				"username STRING," +
				"action STRING," +
				"create_time TIMESTAMP) WITH (" +
				"'connector' = 'mysql-cdc'," +
				"'hostname' = 'under.azure'," +
				"'database-name' = 'flink-test'," +
				"'table-name' = 't_logs'," +
				"'username' = 'root'," +
				"'password' = '123456'" +
				")");

		tbEnv.sqlQuery("select * from t_logs").execute().print();

		// execute program
//		env.execute("Flink Streaming Java API Skeleton");
	}
}
